package net.bwie.realtime.jtp.dws.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.DateTimeUtil;
import net.bwie.realtime.jtp.dws.log.bean.PageViewBean;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;

public class PageViewBeanMapFunction extends RichMapFunction<String, PageViewBean> {

    // 声明变量
    private transient ValueState<String> lastVisitDateState = null;

    // 初始化
    @Override
    public void open(Configuration parameters) throws Exception {
        lastVisitDateState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("last-visit-date", String.class)
        );
    }

    @Override
    public PageViewBean map(String value) throws Exception {

        // 解析 json
        JSONObject jsonObject = JSON.parseObject(value);
        JSONObject object = jsonObject.getJSONObject("common");
        JSONObject pageObject = jsonObject.getJSONObject("page");

        // 根据 地区编码 获取 省份名称
//        String arValue = object.getString("ar");
//        String province = areaDic.get(arValue);

        // uv访客数
        Long uvCount = 0L;

        // 获取上一次访问日期
        String lastVisitDate = lastVisitDateState.value();
        //获取当前日期
        String visitDate = DateTimeUtil.convertLongToString(jsonObject.getLong("ts"), DateTimeUtil.DATE_FORMAT);

        // 判断是否是今天 第一次访问
        if (null == lastVisitDate || !lastVisitDate.equals(visitDate)) {
            uvCount = 1L;
            lastVisitDateState.update(visitDate);
        }
        // 返回实例对象
        return new PageViewBean(
                null,
                null,
                object.getString("ba"),
                object.getString("ch"),
                object.getString("ar"),
                object.getString("isNew"),

                pageObject.get("last_page_id")==null ? 1L : 0L,
                1L,
                pageObject.getLong("during_time"),
                uvCount,

                jsonObject.getLong("ts")

        );
    }

    @Override
    public void close() throws Exception {
        super.close();
    }


}
